package cn.itcast.tags.utils

import cn.itcast.tags.config.ModelConfig
import com.typesafe.config.{Config, ConfigFactory, ConfigValue}
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession

import java.util
import java.util.Map


/**
 * 创建sparkSession对象工具类
 */

object SparkUtils {

  /**
   * 加载Spark Application默认配置文件，设置到SparkConf中
   * @param resource 资源配置文件名称
   * @return SparkConf对象
   */
  def loadConf(resource: String): SparkConf = {
    // 1. 创建SparkConf 对象
    val sparkConf = new SparkConf()
    // 2. 使用ConfigFactory加载配置文件
    val config: Config = ConfigFactory.load(resource)
    // 3. 获取加载配置信息
    val entrySet: util.Set[util.Map.Entry[String, ConfigValue]] = config.entrySet()
    // 4. 循环遍历设置属性值到SparkConf中
    import scala.collection.JavaConverters._
    entrySet.asScala.foreach { entry =>
      // 获取属性来源的文件名称
      val resourceName = entry.getValue.origin().resource()
      if (resource.equals(resourceName)) {
        sparkConf.set(entry.getKey, entry.getValue.unwrapped().toString)
      }
    }
    // 5. 返回SparkConf对象
    sparkConf
  }

  /**
   * 创建SparkSession实例对象，如果是本地模式，设置master
   * @param clazz
   * @param isHive
   */
  def createSparkSession(clazz:Class[_],isHive:Boolean=false): SparkSession ={
    val aparkConf: SparkConf = loadConf("spark.properties")

    //判断是否是本地莫斯
    if(ModelConfig.APP_IS_LOCAL){
      aparkConf.setMaster(ModelConfig.APP_SPARK_MASTER)
    }

    //3.创建SparkSession.Builder对象
    var builder: SparkSession.Builder = SparkSession.builder()
      .appName(clazz.getSimpleName.stripSuffix("&"))
      .config(aparkConf)

    //4.判断是否集成hive,如果集成，设置hive metastore地址
    //如果在config.properties中设置成Hive，表示所有Application都集成hive,否则判断isHive,表示针对某个具体应用
    if(ModelConfig.APP_IS_HIVE || isHive){
      builder = builder
        .config("hive.metastore.uris", ModelConfig.APP_HIVE_META_STORE_URL)
        .enableHiveSupport()
    }

    //5、采用建造者设计模式构建SparkSession实例对象
    val spark: SparkSession = builder.getOrCreate()

    spark
  }

}
